package com.catmiao.spark.rdd.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: RDD_Operator_01_transform
 * @projectName spark_study
 * @description: TODO
 * @author ChengMiao
 * @date 2024/2/2 11:48
 */
object RDD_Operator_17_transform {

  def main(args: Array[String]): Unit = {
    val sparkCon = new SparkConf().setMaster("local[*]").setAppName("rdd")

    val sparkContext = new SparkContext(sparkCon)


    val rdd = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)), 2)


    /**
     * combineByKey
     * param1: 将相同key的第一个数据进行结构的转换，实现操作
     * param2: 分区内的计算规则
     * param3: 分区间的计算规则
     */
    val rsRdd: RDD[(String, (Int, Int))] = rdd.combineByKey(
      v => (v, 1),
      (x: (Int, Int), y) => {
        (x._1 + y, x._2 + 1)
      },
      (x: (Int, Int), y: (Int, Int)) => {
        (x._1 + y._1, x._2 + y._2)
      }
    )

    rsRdd.mapValues{
      case (x,y) => {
        x/y
      }
    }.collect().foreach(println)


    sparkContext.stop()
  }

}
